package com.xf.day07_graphx
import org.apache.log4j.{Level,Logger}
import org.apache.spark._
import org.apache.spark.graphx.{GraphLoader,PartitionStrategy}
/**
 * graphx作为spark的图计算组件，提供了丰富的图操作接口，以及常用的一些算法
 *
 * 三角形计数算法: 统计每个顶点所在的三角形的个数
 */
object SimpleGraphX {
  def main(args: Array[String]) {

    //屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")

    println("===========================================>")
    // Load the edges in canonical order and partition the graph for triangle count
    // 加载数据源
    /**
     *  1.重点解释 canonicalOrientation 参数： ===> 表示的是 控制边的方向是否被“标准化”
     *    如果 canonicalOrientation = true, 系统会把每条边的两个顶点按从小到大排序，强制让边从小 ID 指向大 ID。
     *  2.partitionBy(PartitionStrategy.RandomVertexCut) : 随机将边分配到不同分区。每条边根据其两个顶点 ID 的哈希值随机切分。优点是负载均衡好，适合大多数图算法。
     */
    val graph = GraphLoader
      .edgeListFile(sc, "file:///D:\\workRecord\\spark_project20250924\\data\\SimpleGraphX\\followers.txt", true)
      .partitionBy(PartitionStrategy.RandomVertexCut)

    // Find the triangle count for each vertex
    // 对图进行三角形计数
    val triCounts = graph.triangleCount().vertices

    /**
     * 三角形计数的结果
     *    (4,0)
     *    (1,0)
     *    (6,1)
     *    (3,1)
     *    (7,1)
     *    (2,0)
     */
    println("===========================================>")
    triCounts.collect().foreach( println )
    println("===========================================>")

    // Join the triangle counts with the usernames
    // 读取用户信息
    val users = sc.textFile("file:///D:\\workRecord\\spark_project20250924\\data\\SimpleGraphX\\users.txt")
      .map { line =>
      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }

    /**
     * 用户的信息为:
     *    (1,BarackObama)
     *    (2,ladygaga)
     *    (3,jeresig)
     *    (4,justinbieber)
     *    (6,matei_zaharia)
     *    (7,odersky)
     *    (8,anonsys)
     */
    users.collect().foreach(println)
    println("===========================================>")

    val triCountByUsername = users.join(triCounts).map { case (id, (username, tc)) =>
      (username, tc)
    }

    // Print the result
    /**
     * 最后打印的结果为:
     *    (justinbieber,0)
     *    (BarackObama,0)
     *    (matei_zaharia,1)
     *    (jeresig,1)
     *    (odersky,1)
     *    (ladygaga,0)
     */
    println(triCountByUsername.collect().mkString("\n"))

  }
}